// @Author miaoqing
// @Date 2024/02/22 11:31:00
// @Desc
package network

import (
	"core/component/function"
	"core/component/logger"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"reflect"
	"runtime/debug"
	"sync"
	"sync/atomic"
	"time"
)

type IConn interface {
	OnMessage([]byte)

	Run()

	Close()

	OnClose() //连接断开处理对应逻辑

	OnConnect()

	// 协议解析
	Marshal(msgID uint16, msg interface{}) ([]byte, error)
}

type NetConn struct {
	Conn        net.Conn
	signal      chan interface{}
	lastSignal  chan interface{}
	closed      int32
	stoped      chan bool
	Impl        IConn
	msgBuffer   *MsgBuffer
	msgBuffSize int64
	wgRW        sync.WaitGroup //读写等待队列
	ConnID      int64
}

func NewNetConn(conn net.Conn, msgBuffSize int64) *NetConn {
	tcpConn := &NetConn{
		msgBuffSize: msgBuffSize,
		Conn:        conn,
		msgBuffer:   newMsgBuffer(),
		closed:      -1,
		stoped:      make(chan bool, 1),
		signal:      make(chan interface{}, msgBuffSize),
		lastSignal:  make(chan interface{}, 1),
	}
	return tcpConn
}

// Run tcp连接启动
func (nc *NetConn) Run() {
	if atomic.CompareAndSwapInt32(&nc.closed, -1, 0) {
		nc.wgRW.Add(1)
		go nc.HandleRead()
		nc.wgRW.Add(1)
		go nc.HandleWrite()
		logger.Infof("开启NetCoon localAddr:%v  remoteAddr:%v", nc.Conn.LocalAddr(), nc.Conn.RemoteAddr())
	}
	///timeout := time.NewTimer(time.Second * timeoutTime)
L:
	for {
		select {
		// 等待通到返回 返回后检查连接是否验证完成 如果没有验证 则关闭连接
		//	case <-timeout.C:
		//if !tcper.Verified() {
		//	logger.Errorf("[连接] 验证超时 ip addr %s", tcper.RemoteAddr())
		//	tcper.Close()
		//	break L
		//}
		case <-nc.stoped:
			break L
		}
	}
	//timeout.Stop()
	// block here wait read write goroutine Done
	nc.wgRW.Wait()
	logger.Info("连接相关的goroutine已全部成功退出")
	// 这个时候连接相关的goroutine才全部成功退出 触发OnClose函数
	nc.Impl.OnClose()
}

// HandleRead 处理接收消息的goroutine
func (nc *NetConn) HandleRead() {
	defer func() {
		if err := recover(); err != nil {
			function.WritePanic(string(debug.Stack()), err)
		}
	}()
	defer nc.Close()
	defer nc.wgRW.Done()
	for {
		bDatas, err := nc.msgBuffer.Read(nc)
		if err != nil || err == io.EOF {
			return
		}
		for _, data := range bDatas {
			nc.Impl.OnMessage(data)
		}
	}
}

func (nc *NetConn) HandleWrite() {
	defer func() {
		if err := recover(); err != nil {
			function.WritePanic(string(debug.Stack()), err)
			return
		}
	}()
	defer nc.Conn.Close()
	defer nc.wgRW.Done()
	for {
		select {
		case signal := <-nc.signal:
			buf, ok := signal.([]byte)
			if !ok {
				return
			}
			//buf, err := Encode(data)
			//if err != nil {
			//	return
			//}
			err := nc.msgBuffer.Write(nc, buf...)
			if err != nil {
				return
			}
		case signal := <-nc.lastSignal: // 最后一个通知消息
			data, ok := signal.([]byte)
			if !ok {
				logger.Errorf("write message %v error: msg is not bytes", reflect.TypeOf(signal))
				return
			}
			err := nc.msgBuffer.Write(nc, data...)
			if err != nil {
				logger.Errorf("write message %v error: %v", reflect.TypeOf(signal), err)
				return
			}
			time.Sleep(2 * time.Second)
			return
		case <-nc.stoped:
			logger.Infof(" 关闭conn写入连接 Addr %v", nc.Conn.LocalAddr())
			return
		}
	}

}

func (nc *NetConn) Close() {
	if atomic.CompareAndSwapInt32(&nc.closed, 0, 1) {
		nc.Conn.Close()
		close(nc.stoped)
	}
}

func (nc *NetConn) Read(b []byte) (int, error) {
	return nc.Conn.Read(b)
}

func (nc *NetConn) Write(b []byte) (int, error) {
	return nc.Conn.Write(b)
}

// RemoteAddr 获取远程地址
func (nc *NetConn) RemoteAddr() net.Addr {
	return nc.Conn.RemoteAddr()
}

// AsyncSend 缓存在发送队列里等待发送goroutine取出
func (nc *NetConn) AsyncSend(msgID uint16, msg interface{}) bool {
	if nc.IsShutdown() {
		return false
	}
	data, err := nc.Impl.Marshal(msgID, msg)
	if err != nil {
		logger.Errorf("Marshal msgID:%v and msg to bytes error:%v", msgID, err)
		return false
	}
	// 发送的包体过长
	if uint32(len(data)) > nc.msgBuffer.maxMsgLen {
		logger.Errorf("[发包] 发送的消息包体过长 msgID:%v", msgID)
		return false
	}
	err = nc.Signal(data)
	if err != nil {
		nc.Close() // 向消息通知chan发送失败后关闭连接
		logger.Error(err)
		return false
	}
	return true
}

// AsyncSendRowMsg 发送原始字节切片数据
func (nc *NetConn) AsyncSendRowMsg(data []byte) bool {

	if nc.IsShutdown() {
		return false
	}

	//// 发送的包体过长
	if uint32(len(data)) > nc.msgBuffer.maxMsgLen {
		logger.Error("[发包] 发送的消息包体过长 AsyncSendRowMsg")
		return false
	}

	err := nc.Signal(data)
	if err != nil {
		nc.Close() // 向消息通知chan发送失败后关闭连接
		logger.Error(err)
		return false
	}

	return true
}

// Signal 发送消息同步
func (nc *NetConn) Signal(signal []byte) error {
	select {
	case nc.signal <- signal:
		return nil
	default:
		return nil
	}
}

// LocalAddr 本地地址
func (nc *NetConn) LocalAddr() net.Addr {
	return nc.Conn.LocalAddr()
}

// IsShutdown 连接是否已经主动关闭
func (nc *NetConn) IsShutdown() bool {
	return atomic.LoadInt32(&nc.closed) == 1
}

// AsyncSendLastPacket 缓存在发送队列里等待发送goroutine取出 (发送最后一个消息 发送会关闭tcp连接 终止tcp goroutine)
func (nc *NetConn) AsyncSendLastPacket(msgID uint16, msg interface{}) bool {
	data, err := nc.Impl.Marshal(msgID, msg)
	if err != nil {
		logger.Errorf("Marshal msgID:%v and msg to bytes error:%v", msgID, err)
		return false
	}

	// 发送的包体过长
	if uint32(len(data)) > nc.msgBuffer.maxMsgLen {
		logger.Errorf("[发包] 发送的消息包体过长 msgID:%v", msgID)
		return false
	}

	err = nc.LastSignal(data)
	if err != nil {
		nc.Close() // 向消息通知chan发送失败后关闭连接
		logger.Error(err)
		return false
	}

	return true
}

// LastSignal 发送消息同步
func (nc *NetConn) LastSignal(signal []byte) error {
	select {
	case nc.lastSignal <- signal:
		return nil
	default:
		{
			cmd := binary.LittleEndian.Uint16(signal[2:4])
			return fmt.Errorf("buffer full blocking cmd:%v", cmd)
		}
	}
}
